package com.tiancheng.trade.payment.util.queue;

import com.tiancheng.trade.commom.core.utils.JsonUtil;
import com.tiancheng.trade.commom.core.utils.ServiceLocator;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

/**
 * @Author: likailun
 * @Description: 启动延迟队列
 * @Date: create in 2022/4/7 4:14 下午
 */
@Slf4j
@Component
public class RedisDelayQueueRunner implements CommandLineRunner {

    @Autowired
    private RedisDelayQueueUtil redisDelayQueueUtil;

    @Override
    public void run(String... args) throws Exception {
        RedisDelayQueueEnum[] queueEnums = RedisDelayQueueEnum.values();
        /*for (RedisDelayQueueEnum queueEnum : queueEnums) {
            startThread(queueEnum);
        }*/
    }

    public void startThread(RedisDelayQueueEnum queueEnum) {
        RBlockingQueue rBlockingQueue = redisDelayQueueUtil.getDelayQueue(queueEnum.getCode());
        RedisDelayQueueHandle redisDelayQueueHandle = (RedisDelayQueueHandle) ServiceLocator.getBean(queueEnum.getBean());
        //由于此线程需要常驻，可以新建线程，不用交给线程池管理
        Thread thread = new Thread(() -> {
            log.info("启动监听队列线程:{}" + queueEnum.getCode());
            while (true) {
                try {
                    final Object value = rBlockingQueue.take();
                    log.info("监听队列线程，监听名称：{},内容:{}", queueEnum.getCode(), JsonUtil.toJsonString(value));
                    if (value != null) {
                        redisDelayQueueHandle.execute(value);
                    }
                } catch (Exception e) {
                    log.info("监听队列线程错误,", e);
                }
            }
        });
        thread.setName(queueEnum.getCode());
        thread.start();
    }
}
